package com.bigdata.example;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.junit.Test;
import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.InputStream;
import java.net.URI;

public class TestHDFS {
    @Test
    public void a_createNewFile(){//1.给hdfs分布式文件系统中创建新文件并写入一段字符串
        try{
            String dst="hdfs://master:9000/hdfs/hello.txt";
            Configuration conf=new Configuration();//创建HDFS的配置对象
            conf.set("fs.defaultFS","hdfs://master:9000");
            conf.set("fs.hdfs.impl","org.apache.hadoop.hdfs.DistributedFileSystem");
            FileSystem fs=FileSystem.get(URI.create(dst),conf);//获取FileSystem对象
            byte[]  buff="This is a example of output to HDFS  file".getBytes();
            FSDataOutputStream out=fs.create(new Path(dst));//创建dts对应的hello.txt文件，并用输出流打开此文件
            out.write(buff,0,buff.length);
            out.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }


    @Test
    public void b_createFileByCopy() {//2.给hdfs中创建新文件word.txt并将虚拟机中的a.txt中的文字复制进去
        try{
            String src="/home/hadoop/a.txt";
            String dst="hdfs://master:9000/hdfs/word.txt";
            Configuration conf=new Configuration();//创建HDFS的配置对象
            FileSystem fs=FileSystem.get(URI.create(dst),conf);//获取FileSystem对象
            InputStream in=new BufferedInputStream(new FileInputStream(src));//创建输入流读取src对应的a.txt
            FSDataOutputStream out=fs.create(new Path(dst));//用FileSystem对象的create方法创建dst对应的word.txt文件并用输出流打开它
            IOUtils.copyBytes(in,out,4096,true);
            IOUtils.closeStream(in);//关闭输入流(输出流在copyBytes中会自动关闭)
        }catch (Exception e){
            e.printStackTrace();
        }
    }


    @Test
    public void c_appendFile() {//3.将a.txt文件内容追加到hdfs中word.txt文件中。
        try {
            String localSrc = "/home/hadoop/a.txt";
            String dst = "hdfs://master:9000/hdfs/word.txt";
            BufferedInputStream in = new BufferedInputStream(new FileInputStream(localSrc));//创建输入流读取src对应的a.txt
            Configuration conf = new Configuration();//创建HDFS的配置对象
            conf.set("dfs.client.block.write.replace-datanode-on-failure","NEVER");
            conf.setBoolean("dfs.client.block.write.replace-datanode-on-failure.enable",true);
            //true表示启用datanode替换机制：某个datanode出异常可以替换它
            //false表示不启用datanode替换机制：当某个datanode出异常后，客户端会移除异常的datanode
            FileSystem fs = FileSystem.get(URI.create(dst), conf);//获取FileSystem对象
            FSDataOutputStream out=fs.create(new Path(dst));//用FileSystem对象的create方法创建dst对应的word.txt文件并用输出流打开它
            IOUtils.copyBytes(in,out,4096,false);//和上边的复制方法b_createFileByCopy不同的是：多出了两个set，这里的最后一个参数是false(其余代码基本相同)
            IOUtils.closeStream(in);//关闭输入流(输出流在copyBytes中会自动关闭)
        }catch (Exception e){
            e.printStackTrace();
        }
    }


    @Test
    public void d_readFile() {//4.整体性的读取HDFS中的word.txt文件。
        try {
            String uri = "hdfs://master:9000/hdfs/word.txt";
            Configuration conf = new Configuration();//创建HDFS的配置对象
            FileSystem fs = FileSystem.get(URI.create(uri), conf);//获取FileSystem对象
            InputStream in=fs.open(new Path(uri));//FileSystem对象的open()方法用一个输入流打开uri对应的文件,用于读取文件
            IOUtils.copyBytes(in,System.out,4096,false);//false表示用输入流读取HDFS中的文件追加形式打印到标准控制台输出窗口中
            IOUtils.closeStream(in);//关闭输入流(输出流在copyBytes中会自动关闭)
        }catch (Exception e){
            e.printStackTrace();
        }
    }


    @Test
    public void e_randomReadFile() {//5.随机读取word.txt文件(从文件的任意位置读取内容)(本方法实现的是：每次读取256个字节输出)。
        try {
            String uri = "hdfs://master:9000/hdfs/word.txt";
            Configuration conf = new Configuration();//创建HDFS的配置对象
            FileSystem fs = FileSystem.get(URI.create(uri), conf);//获取FileSystem对象
            InputStream in=fs.open(new Path(uri));//FileSystem对象的open()方法用一个输入流打开uri对应的文件,用于读取文件
            byte[]  buffer=new byte[256];
            int byteRead=0;
            while((byteRead=in.read(buffer))>0){//如果从HDFS的word.txt文件能读取到数据
                System.out.write(buffer,0,byteRead);//就将buffer数组中的0到byteRead之间的有效数据写出到控制台窗口
            }
            IOUtils.closeStream(in);//关闭输入流(输出流在copyBytes中会自动关闭)
        }catch (Exception e){
            e.printStackTrace();
        }
    }


    @Test
    public void f_fileExists() {//6.判断文件是否存在于HDFS中
        try {
            String uri = "hdfs://master:9000/hdfs/word.txt";
            Configuration conf = new Configuration();//创建HDFS的配置对象
            FileSystem fs = FileSystem.get(URI.create(uri), conf);//获取FileSystem对象
            if(fs.exists(new Path(uri))){//用FileSystem的exists判断HDFS中是否存在uri对应的文件
                System.out.printf("文件存在");
            }else{
                System.out.printf("文件不存在");
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }


    @Test
    public void g_upLoadFile() {//7.将本地虚拟机中的maven.xml中的文件上传到HDFS文件系统中
        try {
            String src="/home/hadoop/maven.txt";
            String dst = "hdfs://master:9000/";
            Configuration conf = new Configuration();//创建HDFS的配置对象
            FileSystem fs = FileSystem.get(URI.create(dst), conf);//获取FileSystem对象
            fs.copyFromLocalFile(new Path(src),new Path(dst));//将src本地虚拟机中的文件上传到HDFS中的dst路径里
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    @Test
    public void h_downLoadFile(){//8.将HDFS文件系统中的word.txt文件 下载到本地虚拟机的newWord.txt文件中
        try {
            String src = "hdfs://master:9000/";
            String dst="/home/hadoop/maven.txt";
            Configuration conf = new Configuration();//创建HDFS的配置对象
            FileSystem fs = FileSystem.get(URI.create(src), conf);//获取FileSystem对象
            fs.copyToLocalFile(new Path(src),new Path(dst));//将HDFS中src本地虚拟机中的文件上传到的dst路径里
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    @Test
    public void I_createDir(){//9.在HDFS的hdfs目录下创建一个course文件夹
        try {
            //String uri="hdfs://master:9000/hdfs/course";
            String uri="hdfs://master:9000/hdfs2/course2";
            Configuration conf=new Configuration();//创建HDFS的配置对象
            FileSystem fs=FileSystem.get(URI.create(uri), conf);//获取FileSystem对象
            fs.mkdirs(new Path(uri));
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    @Test
    public void J_renameDir(){//10.将HDFS文件系统中的word.txt文件重命名为newWord.txt
        try {
            String oldName="hdfs://master:9000/hdfs/word.txt";
            String newName="hdfs://master:9000/hdfs/newWord.txt";
            Configuration conf=new Configuration();//创建HDFS的配置对象
            FileSystem fs=FileSystem.get(URI.create(oldName), conf);//获取FileSystem对象
            fs.rename( new Path(oldName), new Path(newName));//用FileSystem的rename方法将第一个参数的文件名，修改为第二个参数的文件名
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    @Test
    public void K_fileStatus(){//11.查看HDFS文件系统中某个文件的状态
        try {
            String uri="hdfs://master:9000/hdfs/newWord.txt";
            Configuration conf=new Configuration();//创建HDFS的配置对象
            FileSystem fs=FileSystem.get(URI.create(uri), conf);//获取FileSystem对象
            if(fs.exists(new Path(uri))) {//用FileSystem的exists判断HDFS中是否存在uri对应的文件
                FileStatus fileStatus=fs.getFileStatus(new Path(uri));
                System.out.println("文件路径:"+ fileStatus.getPath());
                System.out.println("块大小:"+ fileStatus.getBlockSize());
                System.out.println("文件所有者:"+ fileStatus.getOwner() + "文件所属组:"+ fileStatus.getGroup());
                System.out.println("文件权限:"+ fileStatus.getPermission());
                System.out.println("文件长度:"+ fileStatus.getLen());
                System.out.println("备份数:"+ fileStatus.getReplication());
                System.out.println("修改时间:"+ fileStatus.getModificationTime());
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    @Test
    public void L_dirStatus(){//12.查看HDFS文件系统中某个文件夹里所有文件的状态
        try {
            String[] uris={"hdfs://master:9000/hdfs","hdfs://master:9000/hdfs2"};
            Configuration conf=new Configuration();//创建HDFS的配置对象
            FileSystem fs=FileSystem.get(URI.create(uris[0]), conf);//获取FileSystem对象
            Path[]  paths=new Path[uris.length];
            for(int i=0;i<paths.length;i++){
                paths[i]=new Path(uris[i]);
            }
            FileStatus[] status=fs.listStatus(paths);
            for(FileStatus fileStatus:status){
                System.out.println("文件路径:"+ fileStatus.getPath());
                System.out.println("块大小:"+ fileStatus.getBlockSize());
                System.out.println("文件所有者:"+ fileStatus.getOwner() + "文件所属组:"+ fileStatus.getGroup());
                System.out.println("文件权限:"+ fileStatus.getPermission());
                System.out.println("文件长度:"+ fileStatus.getLen());
                System.out.println("备份数:"+ fileStatus.getReplication());
                System.out.println("修改时间:"+ fileStatus.getModificationTime());
                System.out.println("--------------------------------------");
                System.out.println();
            }
            Path[] listedPaths=FileUtil.stat2Paths(status);
            for(Path path:listedPaths){
                System.out.println(path);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }



    @Test
    public void L_deleteFile(){//13.删除HDFS的文件或文件夹
        try {
            String uri="hdfs://master:9000/hdfs/newWord.txt";
            Configuration conf=new Configuration();//创建HDFS的配置对象
            FileSystem fs=FileSystem.get(URI.create(uri), conf);//获取FileSystem对象
            fs.delete(new Path(uri), true);
            //这里要注意一下第二个参数，如果第一个参数为目录，那么此时必须为true(表示一个非空目录及其内容才会被删除<false不会被删除>)
            //如果第一个参数为文件，那么true和false都可以。
        }catch (Exception e){
            e.printStackTrace();
        }
    }




}
